package com.kingdee.hackson.service.impl;

import com.kingdee.hackson.dao.IPlatformSchemaDao;
import com.kingdee.hackson.entity.Position;
import com.kingdee.hackson.service.IPlatformSchemeService;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Supplier;
import java.util.stream.Collectors;

@Service
public class PlatformSchemeService implements IPlatformSchemeService {

    @Resource
    private IPlatformSchemaDao platformSchemaDao;

    @Resource
    private InitOrgIdServiceImpl orgIdService;

    private static final ExecutorService pool = Executors.newCachedThreadPool();

    @Override
    public void executeInScheme(int pageNum, int count, Map<String, Object> responseMap) {
        long startTime = System.currentTimeMillis();
        int startNum = (pageNum - 1) * count;
        List<Long> adminOrgIds = calculateCostTime(() -> orgIdService.getHasrightOrgId(), "[in方案]-获取行政组织id");
        List<Position> positions = calculateCostTime(() -> queryPositionByIn(adminOrgIds, startNum, count), "[in方案]-查询岗位，分页20条");
        responseMap.put("costTime", System.currentTimeMillis() - startTime);
        responseMap.put("positions", positions);
    }

    @Override
    public long getInSchemeResultCount(Map<String, Object> responseMap) {
        long startTime = System.currentTimeMillis();
        List<Long> adminOrgIds = calculateCostTime(() -> orgIdService.getHasrightOrgId(), "[in方案]-获取行政组织id");
        long count = calculateCostTime(() -> getPositionCountByIn(adminOrgIds), "[in方案]-查询岗位总数");
        responseMap.put("costTime", System.currentTimeMillis() - startTime);
        return count;
    }


    @Override
    public void executeInByMultiThread(int pageNum, int count, Map<String, Object> responseMap) throws InterruptedException {
        long startTime = System.currentTimeMillis();
        // 将in中的条件切割，每1w条记录一个线程，并发查询，最后汇总结果
        List<Long> adminOrgIds = calculateCostTime(() -> orgIdService.getHasrightOrgId(), "[in多线程方案]-获取行政组织id");
        int size = adminOrgIds.size();
        // task数量
        int taskNum = Math.max(size / 10000, 1);
        // 每个task处理的数据条数
        int handleCount = size / taskNum;
        CountDownLatch endGate = new CountDownLatch(taskNum);
        // 存放查询结果
        List<Position> resultList = Collections.synchronizedList(new ArrayList<>());
        int startNum = (pageNum - 1) * count;

        for (int i = 0; i < taskNum; i++) {
            int startIndex = i * handleCount;
            int endIndex = Math.min(startIndex + handleCount, size);
            List<Long> subList = adminOrgIds.subList(startIndex, endIndex);
            pool.submit(() -> {
                try {
                    List<Position> positions = calculateCostTime(() -> platformSchemaDao.queryPositionByIn(subList, startNum, count),
                            "[in多线程方案]-线程" + Thread.currentThread().getName() + "-查询岗位，分页20条");
                    resultList.addAll(positions);
                } catch (Exception ex) {
                    ex.printStackTrace();
                } finally {
                    endGate.countDown();
                }
            });

            if (endIndex >= size) {
                break;
            }
        }
        endGate.await();
        // 根据id排序，截取前20条数据
        List<Position> responseData = resultList.stream().sorted(Comparator.comparingLong(Position::getFid)).limit(count).collect(Collectors.toList());
        long costTime = System.currentTimeMillis() - startTime;
        System.out.println("[in多线程方案]-查询岗位，分页20条，总耗时" + costTime + "(ms)");
        responseMap.put("costTime", costTime);
        responseMap.put("positions", responseData);
    }

    @Override
    public void executePlatformScheme(int pageNum, int count, Map<String, Object> responseMap) {
        long startTime = System.currentTimeMillis();
        int startNum = (pageNum - 1) * count;
        List<Position> positions = execute(startNum, count);
        String tableName = getTmpTableName();
        // 另起一个线程删除临时表
        calculateCostTime(() -> {
            pool.submit(() -> deleteMemoryTab(tableName));
            return null;
        }, "[平台临时表方案]-删除临时表");
        long costTime = System.currentTimeMillis() - startTime;
        System.out.println("[平台临时表方案]-查询岗位，分页20条，总耗时" + costTime + "(ms)");
        responseMap.put("costTime", costTime);
        responseMap.put("positions", positions);
    }

    @Override
    public void executePlatformSchemeByMultiThread(int pageNum, int count, Map<String, Object> responseMap) {
        long startTime = System.currentTimeMillis();
        int startNum = (pageNum - 1) * count;
        List<Position> positions = executeByMultiThread(startNum, count);
        String tableName = getTmpTableName();
        // 另起一个线程删除临时表
        calculateCostTime(() -> {
            pool.submit(() -> deleteMemoryTab(tableName));
            return null;
        }, "[平台临时表方案-多线程插入]-删除临时表");
        long costTime = System.currentTimeMillis() - startTime;
        System.out.println("[平台临时表方案-多线程插入]-查询岗位，分页20条，总耗时" + costTime + "(ms)");
        responseMap.put("costTime", costTime);
        responseMap.put("positions", positions);
    }

    @Override
    public long getPlatformSchemeCount(Map<String, Object> responseMap) {
        long startTime = System.currentTimeMillis();
        // 查询行政组织id
        List<Long> adminOrgIds = calculateCostTime(() -> orgIdService.getHasrightOrgId(), "[平台临时表方案]-获取行政组织id");
        // 将查询出来的行政组织id存入行政组织临时表
        initTmpTableData(adminOrgIds);
        long count = queryPositionCount();
        String tableName = getTmpTableName();
        // 另起一个线程删除临时表
        calculateCostTime(() -> {
            pool.submit(() -> deleteMemoryTab(tableName));
            return null;
        }, "[平台临时表方案]-删除临时表");
        long costTime = System.currentTimeMillis() - startTime;
        responseMap.put("costTime", costTime);
        return count;
    }

    @Override
    public void deleteTmpTableForException() {
        deleteMemoryTab(getTmpTableName());
    }

    private List<Position> execute(int startNum, int count) {
        // 查询行政组织id
        List<Long> adminOrgIds = calculateCostTime(() -> orgIdService.getHasrightOrgId(), "[平台临时表方案]-获取行政组织id");
        // 将查询出来的行政组织id存入行政组织临时表
        initTmpTableData(adminOrgIds);
        // 将岗位表和临时表通过行政组织ID等值连接起来，获得符合条件的值
        return calculateCostTime(() -> queryPosition(startNum, count), "[平台临时表方案]-查询岗位，分页20条");
    }

    private List<Position> executeByMultiThread(int startNum, int count) {
        // 查询行政组织id
        List<Long> adminOrgIds = calculateCostTime(() -> orgIdService.getHasrightOrgId(), "[平台临时表方案]-获取行政组织id");
        // 将查询出来的行政组织id存入行政组织临时表
        try {
            initTmpTableDataForMultiThread(adminOrgIds);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 将岗位表和临时表通过行政组织ID等值连接起来，获得符合条件的值
        return calculateCostTime(() -> queryPosition(startNum, count), "[平台临时表方案]-查询岗位，分页20条");
    }

    private void initTmpTableData(List<Long> adminOrgIds) {
        String tmpTableName = getTmpTableName();

        // 创建临时表
        calculateCostTime(() -> {
            platformSchemaDao.generateMemoryTab(tmpTableName);
            return null;
        }, "[平台临时表方案]-创建临时表");

        // 批量插入的insert语句可能过长，需要启用多个线程分片插入

        // 批量将行政组织id存入临时表中
        calculateCostTime(() -> {
            platformSchemaDao.batchInsertMemoryTab(tmpTableName, adminOrgIds);
            return null;
        }, "[平台临时表方案]-行政组织id批量插入临时表");

    }

    private void initTmpTableDataForMultiThread(List<Long> adminOrgIds) throws InterruptedException {
        String tmpTableName = getTmpTableName();
        // 创建临时表
        calculateCostTime(() -> {
            platformSchemaDao.generateMemoryTab(tmpTableName);
            return null;
        }, "[平台临时表方案]-创建临时表");

        // 批量插入的insert语句可能过长，需要启用多个线程分片插入，每个线程5w条数据
        int size = adminOrgIds.size();
        int handleCount = Math.min(50000, size);
        int taskNum = Math.max(size / 50000, 1);
        CountDownLatch endGate = new CountDownLatch(taskNum);

        for (int i = 0; i < taskNum; i++) {
            int startIndex = i * handleCount;
            int endIndex = Math.min(startIndex + handleCount, size - 1);

            List<Long> subList = adminOrgIds.subList(startIndex, endIndex);
            pool.submit(() -> {
                try {
                    // 批量将行政组织id存入临时表中
                    calculateCostTime(() -> {
                        platformSchemaDao.batchInsertMemoryTab(tmpTableName, subList);
                        return null;
                    }, "[平台临时表方案]-线程+ " + Thread.currentThread().getName() + "-行政组织id批量插入临时表["+startIndex+"-"+endIndex+"]");
                } catch (Exception ex) {
                    ex.printStackTrace();
                } finally {
                    endGate.countDown();
                }
            });

            if (endIndex >= size) {
                break;
            }
        }
        endGate.await();
    }

    private List<Long> queryAdminOrgIds() {
        return platformSchemaDao.queryAdminOrgIds();
    }

    private List<Position> queryPosition(int startNum, int count) {
        return platformSchemaDao.queryPosition(getTmpTableName(), startNum, count);
    }

    private long queryPositionCount() {
        return platformSchemaDao.queryPositionCount(getTmpTableName());
    }

    private void deleteMemoryTab(String tableName) {
        platformSchemaDao.dropMemoryTab(tableName);
    }

    private List<Position> queryPositionByIn(List<Long> adminOrgIds, int startNum, int count) {
        return platformSchemaDao.queryPositionByIn(adminOrgIds, startNum, count);
    }

    private long getPositionCountByIn(List<Long> adminOrgIds) {
        return platformSchemaDao.queryPositionCountByIn(adminOrgIds);
    }

    private String getTmpTableName() {
        return Thread.currentThread().getId() + "_tmp";
    }

    private <T> T calculateCostTime(Supplier<T> supplier, String msg) {
        long startTime = System.currentTimeMillis();
        T result = supplier.get();
        long costTime = System.currentTimeMillis() - startTime;
        System.out.println(msg + "耗时为：" + costTime + "(ms)");
        return result;
    }
}
